package cluster

import (
	"com.mgface.disobj/metanode/mq/mgfacemq/server"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/hashicorp/memberlist"
	log "github.com/sirupsen/logrus"
	"io/ioutil"
	"math/rand"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

type update struct {
	Action string // add, del
	Data   map[string]interface{}
}

type delegate struct {
	mtx        sync.RWMutex
	items      map[string]interface{}
	broadcasts *memberlist.TransmitLimitedQueue
	serv       *server.Server
}

func (proxy *delegate) NodeMeta(limit int) []byte {
	return []byte{}
}

//获取gossip服务传递过来的数据
func (proxy *delegate) NotifyMsg(data []byte) {
	if len(data) == 0 {
		return
	}

	switch data[0] {
	//判断数据类型为D
	case 'D': // data
		var updates []update
		if err := json.Unmarshal(data[1:], &updates); err != nil {
			return
		}
		proxy.mtx.Lock()
		for _, u := range updates {
			for k, v := range u.Data {
				switch u.Action {
				case "syn":
					proxy.items[k] = v
				}
			}
		}
		//从master同步过来的数据更新到内存中，slave的cache.Datas数据类型会失真
		for _, v := range proxy.items {
			//fmt.Println("k:", k, "v:", v)
			proxy.serv.AssignThis(v)
		}
		proxy.mtx.Unlock()
	}
}

func (proxy *delegate) GetBroadcasts(overhead, limit int) [][]byte {
	return proxy.broadcasts.GetBroadcasts(overhead, limit)
}

func (proxy *delegate) LocalState(join bool) []byte {
	proxy.mtx.RLock()
	m := proxy.items
	proxy.mtx.RUnlock()
	data, _ := json.Marshal(m)
	return data
}

func (proxy *delegate) MergeRemoteState(buf []byte, join bool) {
	if len(buf) == 0 {
		return
	}
	if !join {
		return
	}
	var m map[string]string
	if err := json.Unmarshal(buf, &m); err != nil {
		return
	}
	proxy.mtx.Lock()
	for k, v := range m {
		proxy.items[k] = v
	}
	proxy.mtx.Unlock()
}

//节点进行排序
type WrapMemberlistNodes []*memberlist.Node

func (s WrapMemberlistNodes) Len() int {
	return len(s)
}

func (s WrapMemberlistNodes) Swap(i, j int) {
	s[i], s[j] = s[j], s[i]
}

func (s WrapMemberlistNodes) Less(i, j int) bool {
	nodenamei := strings.Split(s[i].Name, "-")
	nodei, _ := strconv.ParseInt(nodenamei[0], 10, 64)
	nodenamej := strings.Split(s[j].Name, "-")
	nodej, _ := strconv.ParseInt(nodenamej[0], 10, 64)
	return nodei < nodej
}

func GossipDect(naddr, cluster, gaddr string, serv *server.Server) (*memberlist.Memberlist, error) {
	//hostname, _ := os.Hostname()
	conf := memberlist.DefaultLANConfig()
	proxy := &delegate{
		mtx:   sync.RWMutex{},
		items: make(map[string]interface{}),
		serv:  serv,
	}
	conf.Delegate = proxy
	//以节点启动的纳秒数作为节点名称
	nodename := strconv.FormatInt(time.Now().UnixNano(), 10)
	nodename = fmt.Sprintf("%s-%s", nodename, naddr)
	//节点名称为：节点创建的纳秒数-节点的addr值
	log.Debug("当前节点名称:", nodename)
	conf.Name = nodename
	splitInfo := strings.Split(gaddr, ":")
	if len(splitInfo) != 2 {
		return nil, errors.New("元数据IP地址格式为IP:Port")
	}
	//截取IP地址
	conf.BindAddr = splitInfo[0]
	//截取PORT端口
	conf.BindPort, _ = strconv.Atoi(splitInfo[1])
	conf.LogOutput = ioutil.Discard
	list, err := memberlist.Create(conf)
	if err != nil {
		panic("错误创建集群信息: " + err.Error())
	}
	broadcasts := &memberlist.TransmitLimitedQueue{
		NumNodes: func() int {
			return list.NumMembers()
		},
		RetransmitMult: 1, //最大传输次数
	}
	proxy.broadcasts = broadcasts
	//加入存在的集群节点，最少需要指定一个已知的节点信息
	allNodes := strings.Split(cluster, ",")
	list.Join(allNodes)
	if err != nil {
		panic("错误加入集群信息: " + err.Error())
	}

	go func() {
		for {
			log.Debug("@@@@@@@@@当前集群@@@@@@@@@")
			for i, v := range list.Members() {
				log.Debug(fmt.Sprintf("节点(%d)-%s", i, v.Name))
			}
			log.Debug("#######25S~30S刷新#######")
			time.Sleep(time.Duration(5+rand.Intn(5)) * time.Second)
		}
	}()

	//todo 1.如果存在网络分区，会存在多个master,这个需要重新调整代码,要考虑合并多个master数据存储元数据（心跳数据不需要）
	//todo 2.如果重新连接上之后发现自己状态从master变成slave之后，那么需要把数据同步给当前的master
	for {
		//log.Debug("当前节点排序之前:", list.Members())
		//进行排序
		wnode := WrapMemberlistNodes(list.Members())
		sort.Stable(wnode)
		//log.Debug("当前节点排序之后:", wnode)
		//假如它已经是master，不需要重复set master
		if serv.Nodeinfo.DecideMaster() {
			goto breakMaster
		}

		if wnode[0].Name == nodename {
			serv.MutexServ.Lock()
			serv.Nodeinfo.SetMaster()
			log.Debug(fmt.Sprintf("把当前节点[%s]设置为master.", nodename))
			serv.MutexServ.Unlock()
		}
	breakMaster:
		//当前节点是master，那么只有它才能写数据给其他节点
		if serv.Nodeinfo.DecideMaster() {
			//更新自身心跳
			//server.StartMDeartbeat(gaddr, gaddr)
			data, _ := json.Marshal([]*update{
				{
					Action: "syn",
					Data: map[string]interface{}{
						"memdata": serv.ThisToJson(),
					},
				},
			})

			//广播数据
			broadcasts.QueueBroadcast(&broadcast{
				msg:    append([]byte("D"), data...),
				notify: nil,
			})
		}
		log.Debug("当前节点状态:", serv.Nodeinfo.NodeFlag)

		//----进行心跳注册----
		//取排序之后第一个节点为master
		//wnode[0].Name数据格式为:节点创建的纳秒数-节点的addr值
		master := strings.Split(wnode[0].Name, "-")[1]
		log.Debug("master::", master)
		server.StartMDeartbeat(master, naddr, serv.Nodeinfo.NodeFlag)
		//----心跳注册完成----
		time.Sleep(time.Duration(500+rand.Intn(250)) * time.Millisecond)
	}
}
